package com.wty.zookeeper;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import javax.sql.rowset.FilteredRowSet;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class CuratorDemo {
    public CuratorFramework client = null;
    public static final String zkPath = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";
    private CountDownLatch countDownLatch=new CountDownLatch(1);


    public CuratorDemo() {
        /**
         * 同步创建zk示例，原生api是异步的
         *
         * curator链接zookeeper的策略:ExponentialBackoffRetry
         * baseSleepTimeMs：初始sleep的时间
         * maxRetries：最大重试次数
         * maxSleepMs：最大重试时间
         */
//		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);

        /**
         * curator链接zookeeper的策略:RetryNTimes
         * n：重试的次数
         * sleepMsBetweenRetries：每次重试间隔的时间
         */
        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);

        /**
         * curator链接zookeeper的策略:RetryOneTime
         * sleepMsBetweenRetry:每次重试间隔的时间
         */
//		RetryPolicy retryPolicy2 = new RetryOneTime(3000);

        /**
         * 永远重试，不推荐使用
         */
//		RetryPolicy retryPolicy3 = new RetryForever(retryIntervalMs)

        /**
         * curator链接zookeeper的策略:RetryUntilElapsed
         * maxElapsedTimeMs:最大重试时间
         * sleepMsBetweenRetries:每次重试间隔
         * 重试时间超过maxElapsedTimeMs后，就不再重试
         */
//		RetryPolicy retryPolicy4 = new RetryUntilElapsed(2000, 3000);

        client = CuratorFrameworkFactory.builder().connectString(zkPath).sessionTimeoutMs(20000).
                retryPolicy(retryPolicy).namespace("wty").build();
        client.start();

    }

    public void closeClient() {
        if (client != null) {
            client.close();
        }
    }

    public static void main(String args[]) throws Exception {
        CuratorDemo curatorDemo = new CuratorDemo();

//        创建
//        String nodePath = "/first/test1";
//        byte[] bytes="helloworld".getBytes();
//        curatorDemo.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
//                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(nodePath,bytes);
//
//        更新节点数据 默认就是更新最新版的version 也可以withversion
//        byte[] newBytes="hellowty123".getBytes();
//        curatorDemo.client.setData().forPath(nodePath,newBytes);
//
//        删除节点数据
//        String deletePath="/second";
//        curatorDemo.client.delete()
////                .guaranteed() //如果删除失败，删除命令会一直在后端进行，直到删除成功
//                .deletingChildrenIfNeeded()
//                .forPath(deletePath);
//
//        读取节点数据
//        Stat stat=new Stat();
//        byte[] data=curatorDemo.client.getData().storingStatIn(stat).forPath(nodePath);
//        System.out.println("节点数据"+new String(data));
//        System.out.println("节点版本号"+stat.getVersion());
//
//        查询子节点
//        String superPath="/first";
//        List<String> childList=curatorDemo.client.getChildren().forPath(superPath);
//        childList.stream().forEach(System.out::println);
//
//        查询是否为空
//        String queryPath="/first/test5";
//        Stat stat=curatorDemo.client.checkExists().forPath(queryPath);
//        System.out.println(stat);
//
//        添加watcher，该watcher仅仅作用一次
//        curatorDemo.client.getData().usingWatcher(new MyWatcher()).forPath(nodePath);
//        String path1="/first/watcher1";
//        curatorDemo.client.create().creatingParentsIfNeeded().forPath(path1, "watcher1".getBytes());
//        curatorDemo.client.getData().usingWatcher(new MyWatcher()).forPath(path1);
//        curatorDemo.client.getData().usingWatcher(new MyCuratorWatcher()).forPath(path1);
//        curatorDemo.client.setData().forPath(path1,"changed".getBytes());
//        curatorDemo.client.setData().forPath(path1,"changedd".getBytes());
//       curatorDemo.client.delete().deletingChildrenIfNeeded().forPath(path1);
//
//
//        //节点的监听(增删改 NodeCreated NodeDataChanged NodeDeleted)
//        String listenPath="/first/test1";
//        final NodeCache nodeCache=new NodeCache(curatorDemo.client,listenPath);
//        nodeCache.start(true);
//        if(nodeCache.getCurrentData()!=null){
//            System.out.println("节点初始化数据为"+new String(nodeCache.getCurrentData().getData()));
//        }else {
//            System.out.println("节点数据为空");
//        }
//        nodeCache.getListenable().addListener(new NodeCacheListener() {
//            @Override
//            public void nodeChanged() throws Exception {
//                if(nodeCache.getCurrentData()==null){
//                    System.out.println("节点数据为空");
//                    return;
//                }
//                String data=new String(nodeCache.getCurrentData().getData());
//                System.out.println("节点路径"+nodeCache.getCurrentData().getPath()+"节点数据"+data);
//            }
//        });
//        Thread.sleep(100000);
//        curatorDemo.client.setData().forPath(listenPath,"1".getBytes());
//        Thread.sleep(2000);
//        curatorDemo.client.setData().forPath(listenPath,"2".getBytes());
//        Thread.sleep(2000);
//        curatorDemo.client.setData().forPath(listenPath,"3".getBytes());
//        Thread.sleep(2000);
//        curatorDemo.client.setData().forPath(listenPath,"4".getBytes());
//        Thread.sleep(2000);
//        curatorDemo.client.setData().forPath(listenPath,"5".getBytes());
//        Thread.sleep(2000);


        //子节点监听
        String childNodePath="/first";
        //cache 设置节点数据的缓存状态
        PathChildrenCache childrenCache=new PathChildrenCache(curatorDemo.client,childNodePath,true);
        /**
         * StartMode: 初始化方式
         * POST_INITIALIZED_EVENT：异步初始化，初始化之后会触发事件 （因为异步，挂起线程才打印值）
         * NORMAL：异步初始化
         * BUILD_INITIAL_CACHE：同步初始化 （立马打印出子节点的值）
         * 异步会立马触发一次监听
         */
        //childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        //childrenCache.start(PathChildrenCache.StartMode.NORMAL);
        List<ChildData> childDataList =childrenCache.getCurrentData();
        System.out.println("当前节点的数据列表为");
        for (ChildData child:childDataList){
            System.out.println(child.getPath());
        }

        //添加监听
        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                if(event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)){
                    /**
                     * POST_INITIALIZED_EVENT 初始化时会触发这个操作
                     */
                    System.out.println("子节点初始化");
                }
                else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)){
                    System.out.println("新增子节点路径："+event.getData().getPath()+"  新增子节点值:"+new String(event.getData().getData()));
                }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
                    System.out.println("更新子节点路径："+event.getData().getPath()+"  子节点值:"+new String(event.getData().getData()));
                }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                    System.out.println("删除子节点路径："+event.getData().getPath()+"  更新后子节点值:"+new String(event.getData().getData()));
                }
            }
        });



        curatorDemo.countDownLatch.await();






//        Thread.sleep(10000);

    }

}
